package com.atbeijing.D07;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 保存检查点,以及触发器使用
 * 每次调用触发器都会产生一个TriggerResult来决定窗口接下来发生什么
 *
 * FIRE：如果window operator有ProcessWindowFunction这个参数，将会调用这个ProcessWindowFunction。
 * 如果窗口仅有增量聚合函数(ReduceFunction或者AggregateFunction)作为参数，那么当前的聚合结果将会被发送。窗口的state不变
 */
public class Example1 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //flink默认只保存最近一次的状态(检查点)
        //每隔10秒状态从内存向状态后端保存一次
        env.enableCheckpointing(10*1000);
        //配置状态后端的文件路径
        env.setStateBackend(new FsStateBackend("file:///C:/tmp"));

        env
                .addSource(new SensorSource())
                .filter(r -> r.id.equals("sensor_1"))
                .keyBy(r -> r.id)
                //每10秒一个窗口
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                //每个窗口一个触发器
                .trigger(new Trigger<SensorReading, TimeWindow>() {
                   //窗口中每个元素执行一次
                    @Override
                    public TriggerResult onElement(SensorReading element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
                        //这里是单例的所以才有if判断,每个元素来都new,但是返回的是同一个实例
                        ValueState<Boolean> firstUpdate = ctx.getPartitionedState(
                                new ValueStateDescriptor<Boolean>("first-update", Types.BOOLEAN)
                        );
                        //如果是窗口的第一个元素
                        if (firstUpdate.value()==null){
                            System.out.println("窗口第一个元素到达"+ctx.getCurrentProcessingTime());
                            //下个整数秒时间错
                            long ts = ctx.getCurrentProcessingTime() + (1000 - ctx.getCurrentProcessingTime() % 1000L);
                            //设置处理定时器
                            ctx.registerProcessingTimeTimer(ts);
                            firstUpdate.update(true);
                        }
                        return TriggerResult.CONTINUE;
                    }

                    //处理定时器触发时调用
                    @Override
                    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        long ts = ctx.getCurrentProcessingTime() + (1000 - ctx.getCurrentProcessingTime() % 1000L);
                        //如果下一秒小于窗口结束时间
                        if (ts < window.getEnd()){
                            ctx.registerProcessingTimeTimer(ts);
                        }
                        //如果window operator有ProcessWindowFunction这个参数，将会调用这个ProcessWindowFunction。
                        return TriggerResult.FIRE;
                    }

                    //事件定时器触发时调用
                    @Override
                    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
                        ValueState<Boolean> firstUpdated = ctx.getPartitionedState(
                                new ValueStateDescriptor<Boolean>("first-update", Types.BOOLEAN)
                        );
                        firstUpdated.clear();
                    }
                })
                .aggregate(new Agg(), new WindowResult())
                .print();

        env.execute();
    }

    public static class Agg implements AggregateFunction<SensorReading, Tuple2<Double, Long>, Double> {
        @Override
        public Tuple2<Double, Long> createAccumulator() {
            return Tuple2.of(0.0, 0L);
        }

        @Override
        public Tuple2<Double, Long> add(SensorReading value, Tuple2<Double, Long> accumulator) {
            return Tuple2.of(accumulator.f0 + value.temperature, accumulator.f1 + 1L);
        }

        @Override
        public Double getResult(Tuple2<Double, Long> accumulator) {
            return accumulator.f0 / accumulator.f1;
        }

        @Override
        public Tuple2<Double, Long> merge(Tuple2<Double, Long> a, Tuple2<Double, Long> b) {
            return null;
        }
    }

    public static class WindowResult extends ProcessWindowFunction<Double, String, String, TimeWindow> {
        @Override
        public void process(String s, Context context, Iterable<Double> elements, Collector<String> out) throws Exception {
            out.collect("窗口的平均值是这个：" + elements.iterator().next());
        }
    }

}
